package org.atguigu.gmall.realtime.app.dwd.db;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.atguigu.gmall.realtime.app.BaseSQLApp;
import org.atguigu.gmall.realtime.common.Constant;
import org.atguigu.gmall.realtime.util.SQLUtil;

/**
 * Project:gmallRealTime
 * Package:org.atguigu.gmall.realtime.app.dwd.db
 * Author: ZengHaiFeng
 * CreateTime:2023/4/25 20:55:48
 * Description: Todo
 * Version：1.0
 */
public class Dwd_06_DwdTradePayDetailSuc extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_06_DwdTradePayDetailSuc().init(30006,2,"Dwd_06_DwdTradePayDetailSuc");
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tbe) {
        // read data from dwd_trade_order_detail
        tbe.executeSql("create table `dwd_trade_order_detail`(" +
                " id string, " +
                " order_id string, " +
                " user_id string, " +
                " sku_id string, " +
                " sku_name string, " +
                " province_id string, " +
                " activity_id string, " +
                " activity_rule_id string, " +
                " coupon_id string, " +
                " date_id string, " +
                " create_time string, " +
                " sku_num string, " +
                " split_original_amount string, " +
                " split_activity_amount string, " +
                " split_coupon_amount string, " +
                " split_total_amount string, " +
                " ts bigint , " +
                " et as to_timestamp_ltz(ts,0), " +
                " watermark for et as et - interval '3' second ) " +
                SQLUtil.getKafkaSourceDDL(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL,
                        "Dwd_05_DwdTradeOrderCancelDetail"));

        // read data from ods_db
        readOdsDb(tbe,"Dwd_06_ods_db");

        // read dictionary table
        readBaseDic(tbe);

        // filter payment successful data
        Table payment_suc = tbe.sqlQuery("select" +
                " `data`['user_id'] user_id, " +
                " `data`['order_id'] order_id, " +
                " `data`['payment_type'] payment_type, " +
                " `data`['callback_time'] payment_time, " +
                " `pt`, " +
                " `et`, " +
                " `ts` " +
                " from ods_db where " +
                " `database`='gmall2023' and " +
                " `table`='payment_info' and " +
                " `type`='update' and " +
                " `old`['payment_status']='1601' and " +
                " `data`['payment_status']='1602' ");
        tbe.createTemporaryView("payment_suc",payment_suc);
        // there are three tables join
        Table result = tbe.sqlQuery(" select " +
                " od.id order_detail_id, " +
                " od.order_id order_id, " +
                " od.user_id, " +
                " od.sku_id, " +
                " od.sku_name, " +
                " od.province_id, " +
                " od.activity_id, " +
                " od.activity_rule_id, " +
                " od.coupon_id, " +
                " ps.payment_type payment_type_code, " +
                " dc.dic_name payment_type_name, " +
                " ps.payment_time, " +
                " od.sku_num, " +
                " od.split_original_amount, " +
                " od.split_activity_amount, " +
                " od.split_coupon_amount, " +
                " od.split_total_amount split_payment_amount, " +
                " ps.ts " +
                " from payment_suc as ps join dwd_trade_order_detail as od " +
                " on ps.order_id=od.order_id " +
                " and od.et >= ps.et - interval '30' minute  and od.et <= ps.et + interval '5' second " +
                " join base_dic dc on ps.payment_type=dc.dic_code");

        // output the final result into kafka topic

        tbe.executeSql("create table dwd_trade_pay_detail_suc ( " +
                " order_detail_id string, " +
                " order_id string, " +
                " user_id string, " +
                " sku_id string, " +
                " sku_name string, " +
                " province_id string, " +
                " activity_id string, " +
                " activity_rule_id string, " +
                " coupon_id string, " +
                " payment_type_code string, " +
                " payment_type_name string, " +
                " payment_time string, " +
                " sku_num string, " +
                " split_original_amount string, " +
                " split_activity_amount string, " +
                " split_coupon_amount string, " +
                " split_payment_amount string, " +
                " ts bigint )" + SQLUtil.getKafkaSinkDDL(Constant.TOPIC_DWD_TRADE_PAY_DETAIL_SUC));
        result.executeInsert("dwd_trade_pay_detail_suc");
    }
}
